package com.atguigu.flink.sqlfunction;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2023/2/6
 *
 *      输入1行N列，输出1行1列
 *
 *      举例：把字符串转大写
 */
public class Demo2_ScalarFunction
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = "create table t1( id string,ts bigint ,vc int ) with(" +
            "                    'connector' = 'filesystem' ," +
            "                    'path' = 'data/func/sensor.txt' , " +
            "                    'format' = 'csv'  " +
            "                    )";

        //执行建表(写操作)
        tableEnvironment.executeSql(createTableSql);

        //把t1变为Table对象
        Table table = tableEnvironment.from("t1");

        //1.先创建一个函数对象
        MyUpper myUpper = new MyUpper();

        //使用方式1，在TableAPI中，不注册(给函数起名字)，直接用
        //table.select($("id"),call(myUpper,$("id"))).execute().print();

        //注册函数，再用
        tableEnvironment.createTemporaryFunction("a",myUpper);

        //使用方式2，在TableAPI中，用函数名调用
        //table.select($("id"),call("a",$("id"))).execute().print();

        //使用方式3，在SQL中，用函数名调用
        tableEnvironment.sqlQuery("select id,a(id) from t1").execute().print();


    }

    public static class MyUpper extends ScalarFunction
    {
        /*
            有方法是函数的处理逻辑
                方法名字必须是 eval
                必须有返回值

             @DataTypeHint:加在方法的参数上，声明入参类型，提高编译器编译的速度
         */
        public String eval(@DataTypeHint("STRING") String o) {
            if (o == null){
                return "NULL";
            }else{
                return o.toUpperCase();
            }
        }

    }
}
